﻿#region LicenseHeader

// Copyright 2012 The Trustees of Indiana University.  All rights reserved.
// 
// The Indiana University licenses this file to you under the Apache License, 
// Version 2.0 (the "License"); you may not use this file except in compliance 
// with the License.  You may obtain a copy of the License at 
// 
// http://www.apache.org/licenses/LICENSE-2.0
// 
// Unless required by applicable law or agreed to in writing, software 
// distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
//  WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.  
// See the License for the specific language governing permissions and 
// limitations under the License.
// 
// @author Thilina Gunarathne (tgunarat@indiana.edu)

#endregion

using System;
using System.Collections.Generic;
using System.Threading;
using AzureMRCore.DataModel;
using AzureMRCore.Drivers;
using AzureMRCore.IterativeDataCache;
using AzureMRCore.Utils;
using Microsoft.WindowsAzure;
using Microsoft.WindowsAzure.ServiceRuntime;
using Microsoft.WindowsAzure.StorageClient;

namespace AzureMRCore.MapRed
{
    public class ReduceWorker
    {
        // TODO:  private bool exponentialBackOff;
        private readonly CloudTableClient _cloudTableClient;
        private readonly DataCache _dataCache;
        private readonly MRConf _mrConf;
        private readonly int _reduceQPollingSleepTime;
        private readonly CloudQueue _reduceQueue;
        private readonly int _reduceTimeOut;
        private readonly string _roleId;
        private readonly CloudStorageAccount _storageAccount;
        private readonly string _tableBaseAddress;


        public ReduceWorker(CloudStorageAccount storageAccount, CloudQueue queue, MRConf mrConf, DataCache dataCache)
        {
            _reduceQueue = queue;

            _mrConf = mrConf;
            _reduceQPollingSleepTime = mrConf.RedQPollingInterval;
            _reduceTimeOut = mrConf.ReduceTimeOut;
            _dataCache = dataCache;
            //TODO exponentialBackOff = mrConf.exponentialBackoff;

            _storageAccount = storageAccount;
            _tableBaseAddress = _storageAccount.TableEndpoint.ToString();
            _cloudTableClient = _storageAccount.CreateCloudTableClient();
            _roleId = RoleEnvironment.CurrentRoleInstance.Id;
        }

        public void RunReduce()
        {
            //while items in the specific reduce queue
            while (true)
            {
                try
                {
                    var reduceTimeoutSpan = new TimeSpan(_reduceTimeOut/3600, (_reduceTimeOut%3600)/60,
                                                         (_reduceTimeOut%60));
                    //get the reduce task message
                    CloudQueueMessage msg = _reduceQueue.GetMessage(reduceTimeoutSpan);

                    if (msg != null)
                    {
                        Dictionary<string, object> reduceParams = JSONUtils.parseJSON(msg.AsString);
                        var reduceTaskContext = new ReduceContext(reduceParams, _storageAccount, _cloudTableClient,
                                                                  _tableBaseAddress);
                        MRDriver mrDriver = _mrConf.MRDrivers[reduceTaskContext.AppName];
                        IReducer reducer = mrDriver.GetReducer();

                        reducer.Init(_storageAccount, _mrConf, mrDriver, _dataCache);
                        reduceTaskContext.Attempt = reduceTaskContext.SetStateToExecuting(_roleId);
                        reducer.ExecuteReduce(reduceTaskContext);
                        _reduceQueue.DeleteMessage(msg);
                    }
                    else
                    {
                        Thread.Sleep(_reduceQPollingSleepTime);
                    }
                }
                catch (Exception e)
                {
                    //Trace.TraceError(string.Format("Exception when processing queue item. Message: '{0}'", e.Message));
                    var errorContext = new ErrorDataContext(_tableBaseAddress,
                                                            _storageAccount.Credentials, "AzureMRErrors");

                    _cloudTableClient.CreateTableIfNotExist(errorContext.Name);
                    var error = new ErrorDataModel(e, "reducetask", e.Message);
                    errorContext.AddObject(errorContext.Name, error);
                    errorContext.SaveChangesWithRetries();
                }
            }
        }
    }
}